package com.chencong.online.driver.dwd;

import com.chencong.online.bean.OrderEventBean;
import com.chencong.online.bean.OrderResultBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.OrderEventMapFunc;
import com.chencong.online.function.OrderPatternSelectFunc;
import com.chencong.online.function.OrderPatternTimeoutFunc;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.net.URL;

/**
 * @program: user-behavior-analysis-online
 * @ClassName DwdOrderPayTimeOutDriver
 * @description:超时订单监控(15min内)
 * @author: chencong
 * @create: 2022-01-05 17:19
 **/
public class DwdOrderPayTimeOutDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        env.setParallelism(1);
        //1、source
        URL resource = DwdOrderPayTimeOutDriver.class.getResource("/OrderLog.csv");
        DataStreamSource<String> inputDS = env.readTextFile(resource.getPath());

//        inputDS.print("初始化数据");

        //2、transform
        //将数据格式化并提取时间戳
        SingleOutputStreamOperator<OrderEventBean> orderEventBeanDS = inputDS.map(new OrderEventMapFunc())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderEventBean>forMonotonousTimestamps() //升序
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEventBean>() {
                            @Override
                            public long extractTimestamp(OrderEventBean element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        })
                );
        //使用cep编程
        //1. 定义一个带时间限制的模式
        Pattern<OrderEventBean, OrderEventBean> orderPayPattern = Pattern.
                <OrderEventBean>begin("create").where(new SimpleCondition<OrderEventBean>() {
                    @Override
                    public boolean filter(OrderEventBean value) throws Exception {
                        return "create".equals(value.getEventType());
                    }
                })
                .followedBy("pay").where(new SimpleCondition<OrderEventBean>() {
                    @Override
                    public boolean filter(OrderEventBean value) throws Exception {
                        return "pay".equals(value.getEventType());
                    }
                })
                .within(Time.minutes(15));

        // 2. 定义侧输出流标签，用来表示超时事件
        OutputTag<OrderResultBean> timeOutTag = new OutputTag<OrderResultBean>("order-time-out") {
        };

        // 3. 将pattern应用到输入数据流上，得到pattern stream
        PatternStream<OrderEventBean> patternDS = CEP.pattern(orderEventBeanDS.keyBy(OrderEventBean::getOrderId), orderPayPattern);
        // 4. 调用select方法，实现对匹配复杂事件和超时复杂事件的提取和处理
        SingleOutputStreamOperator<OrderResultBean> selectDS = patternDS.select(timeOutTag, new OrderPatternTimeoutFunc(), new OrderPatternSelectFunc());

        selectDS.print("正常数据");

        selectDS.getSideOutput(timeOutTag).print("超时数据");


        env.execute("DwdOrderPayTimeOutDriver");

    }
}
